/**
 *  Version:     @(#)aio_slot.c    0.2.11 15/06/2007
 *
 *  Authors:     Jinyuan Hei <heijinyuan@hotmail.com>
 *               Hailong Xia <hlxxxx@gmail.com>
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2 of the License, or (at your option) any later version.
 *
 *  This library is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty
 *  of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
 *  the GNU Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General
 *  Public License along with this library; if not, write to the
 *  Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 *  Boston, MA 02111-1307  USA
 */

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <assert.h>

#ifdef _WIN32
# include <winsock2.h>
# include <ws2tcpip.h>
#else
# include <sys/socket.h>
# include <netinet/in.h>
#endif

#include "aio_log.h"
#include "aio_socket.h"
#include "aio_slot.h"

#define TLS_ACCEPT_EXPIRED  10  /* second */
#define TLS_CONNECT_EXPIRED 10  /* second */
#define SLOT_EXCEPTION_TIMEOUT 10 /* millin second */
#define MAX_RECVBUF_LEN AIO_BUF_PLSIZE

static int datagram_recv(struct aio_slot *slot);
static int stream_recv(struct aio_slot *slot);

void 
aio_onclosed(int fd, short events, void *arg)
{
        struct aio_slot *slot;
 
        slot = (struct aio_slot *)arg;
        assert(slot);

        aio_slot_close(slot, 1);
}

static struct aio_slot*
aio_slot_new(struct aio_base *ab, int fd, 
        unsigned long dstip/* network byte order */, 
        unsigned short dstport/* network byte order */)
{
        struct aio_slot *slot;

        assert(ab);

        slot = (struct aio_slot *)calloc(1, sizeof(struct aio_slot));
        assert(slot);

        slot->refcnt = 1; /* init reference count for new aio slot object */
        slot->ab = ab;
        slot->fd = fd;
        slot->dstip = dstip;
        slot->dstport = dstport;

#if 0 /* hl, isn't necessary. */
        if (slot->fd)
                set_socket_nonblocking(fd);
#endif 
        slot->sqcapability = AIO_SLOT_SENDQUE_CAPABILITY;
        slot->sq.tail = &slot->sq;
 
        slot->etv.tv_sec = 0;
        slot->etv.tv_usec = SLOT_EXCEPTION_TIMEOUT;

        aio_evtimer_set(&slot->eev, aio_onclosed, slot);
        aio_evbase_set(&slot->eev, SLOTEBP(slot));

        return slot;
}

#ifdef _AIO_USE_OPENSSL

static void aio_tls_accept(struct aio_slot *slot);

static void
aio_ontlsaccepted(int fd, short events, void *arg)
{
        struct aio_slot *slot;

        slot = (struct aio_slot *)arg;
        assert(slot);

        if (slot->io_errno != 0) {
                log_warn("Tls accept failed.\n");
                if (slot->on_accepted) { 
                        (slot->on_accepted)(slot);
                }
                return;
        }

        aio_tls_accept(slot);
}

static void 
aio_tls_accept(struct aio_slot *slot)
{
        int ret;

        ret = ssl_accept(slot->ssl);
        if (ret > 0) {
                slot->l_connected = 1;
                if (slot->on_accepted) {
                        slot->on_accepted(slot);
                }
        } else if (ret == 0) {
                struct timeval tv;
                tv.tv_sec = TLS_ACCEPT_EXPIRED;
                tv.tv_usec = 0; 
                if (aio_event_add(&slot->rev, &tv) < 0) {
                        log_warn("Failed to call aio_tls_accept.\n");
                        aio_slot_close(slot, 1);
                }
        } else { 
                log_warn("Failed to call aio_tls_accept.\n");
                aio_slot_close(slot, 1);
        }
}
#endif /* _AIO_USE_OPENSSL */

static void
aio_onaccepted(int fd, short events, void *arg)
{
        struct aio_slot *slot, *slot2;
        unsigned long frmip;
        unsigned short frmport;

        slot = (struct aio_slot *)arg;
        assert(slot);

        if (slot->l_close) return;

        if (!(events & AIO_EV_READ)) {
                debugp("Suffer a socket error.\n");
                aio_slot_close(slot, 1);
                return;
        } 

        slot->io_errno = 0;
        fd = tcp_accept(fd, &frmip, &frmport, 0);
        if (fd <= 0) {
                slot->io_errno = errnum;
                if (fd < 0) {
                        aio_slot_close(slot, 1);
                }
                return;
        }

        slot2 = aio_slot_new(slot->ab, fd, frmip, frmport);
        if (slot2) {
                slot2->type = slot->type;
                slot2->on_accepted = slot->on_accepted;
                slot2->usrpointer = slot->usrpointer;
                slot2->frmip = frmip;
                slot2->frmport = frmport;
                LIST_INSERT_AFTER(slot, slot2, entries);
        }

#ifdef _AIO_USE_OPENSSL
        if (slot2->type == AIO_TLS) {
                struct timeval tv;
                slot2->ctx = slot->ctx;
                slot2->ssl = ssl_alloc(slot2->ctx, fd);
                if (slot2->ssl == NULL) {
                        log_warn("Failed to call ssl_alloc.\n");
                        aio_slot_close(slot2, 1);
                        return;
                }

                aio_event_set(&slot2->rev, fd, AIO_EV_READ, aio_ontlsaccepted, slot2);
                aio_evbase_set(&slot2->rev, SLOTEBP(slot2));
                tv.tv_sec = TLS_ACCEPT_EXPIRED;
                tv.tv_usec = 0; 
                if (aio_event_add(&slot2->rev, &tv) < 0) {
                        log_warn("Failed to call aio_tls_accept.\n");
                        aio_slot_close(slot2, 1);
                }
                return;
        }
#endif
        slot2->l_connected = 1;
        if (slot2->on_accepted) {
                slot2->on_accepted(slot2);
        }
}

#ifdef _AIO_USE_OPENSSL

static void aio_tls_connect(struct aio_slot *slot);

static void
aio_ontlsconnected(int fd, short events, void *arg)
{
        struct aio_slot *slot;

        slot = (struct aio_slot *)arg;
        assert(slot);

        if (slot->io_errno != 0) {
                log_warn("Tls connect failed.\n");
                if (slot->on_connected) { 
                        (slot->on_connected)(slot);
                }
                return;
        }

        aio_tls_connect(slot);
}

static void 
aio_tls_connect(struct aio_slot *slot)
{
        int ret;
        
        ret = ssl_connect(slot->ssl);
        if (ret > 0) {
                slot->l_connected = 1;
                if (slot->on_connected) {
                        (slot->on_connected)(slot);
                }
        } else if (ret == 0) {
                struct timeval tv;
                aio_event_set(&slot->rev, slot->fd, AIO_EV_READ, 
                        aio_ontlsconnected, slot);
                aio_evbase_set(&slot->rev, SLOTEBP(slot));
                tv.tv_sec = TLS_CONNECT_EXPIRED;
                tv.tv_usec = 0; 
                if (aio_event_add(&slot->rev, &tv) < 0) {
                        log_warn("Failed to call aio_tls_accept.\n");
                        aio_slot_close(slot, 1);
                }
        } else {
                log_warn("Failed to call ssl_connect.\n");
                aio_slot_close(slot, 1);
        }
}
#endif /* _AIO_USE_OPENSSL */

static void
aio_onconnected(int fd, short events, void *arg)
{
        int ret, error, len;
        struct aio_slot *slot;

        slot = (struct aio_slot *)arg;
        assert(slot);

        if (events & AIO_EV_TIMEOUT) {
                debugp("Operate a timeout.\n");
                slot->io_errno = AIO_EV_TIMEOUT;
                if (slot->on_connected)
                        (slot->on_connected)(slot);
                return;
        } else if (!(events & AIO_EV_WRITE)) {
                debugp("Suffer a socket error.\n");
                aio_slot_close(slot, 1);
                return;
        }

        len = sizeof(error);
        ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, (char*)&error, (socklen_t*)&len);
        if (ret || error) {
                slot->io_errno = errnum;
                aio_slot_close(slot, 1);
                return;
        }

#ifdef _AIO_USE_OPENSSL
        if (slot->type == AIO_TLS) {
                aio_tls_connect(slot);
                return;
        }
#endif /* _AIO_USE_OPENSSL */

        slot->l_connected = 1;
        if (slot->on_connected) {
                (slot->on_connected)(slot);
        }
}

static void
aio_onread(int fd, short events, void *arg)
{
        int ret;
        struct aio_slot *slot;

        slot = (struct aio_slot *)arg;
        assert(slot);

        if (!slot->l_read) return;

        if (!slot->l_permanence)
                slot->l_read = 0;

        if (events & AIO_EV_TIMEOUT) {
                debugp("Aio read expired.\n");
                slot->io_errno = AIO_EV_TIMEOUT;
                if (slot->on_received)
                        (slot->on_received)(slot);
                return;
        } else if (!(events & AIO_EV_READ)) {
                debugp("Suffer a socket error.\n");
                aio_slot_close(slot, 1);
                return;
        } 
					
        if (!slot->rb) {
                slot->rb = AIO_BUF_ALLOC(slot->ab);
                assert(slot->rb);
        }

        if (slot->type & AIO_STREAM) { /* FILE and TCP and TLS */
                ret = stream_recv(slot);
        } else { /* UDP and RAW Sockets */
                ret = datagram_recv(slot);
        }

        if (ret <= 0) {
                if (ret < 0) {
                        aio_slot_close(slot, 1);
                }
                return;
        }

        if (slot->on_received) {
                (slot->on_received)(slot);
        }
}

static void
aio_onwrite(int fd, short events, void *arg)
{
        int ret, nbytes;
        struct aio_slot *slot;

        slot = (struct aio_slot *)arg;
        assert(slot);

       if (events & AIO_EV_TIMEOUT) {
                debugp("Aio read expired.\n");
                slot->io_errno = AIO_EV_TIMEOUT;
                if (slot->on_sent)
                        (slot->on_sent)(slot);
                return;
        } else if (!(events & AIO_EV_WRITE)) {
                debugp("Suffer a socket error.\n");
                aio_slot_close(slot, 1);
                return;
        }

        if (!slot->l_write || !slot->sq.sb)
                return;

        nbytes = BLEN(slot->sq.sb);

        slot->io_errno = 0;

#ifdef _AIO_USE_OPENSSL
        if (slot->type == AIO_TLS && slot->l_connected) {
                ret = ssl_send(slot->ssl, BPTR(slot->sq.sb), nbytes);
        } else
#endif /* _AIO_USE_OPENSSL */
        if (slot->type == AIO_FILE ||
                (slot->type == AIO_TCP && slot->l_connected)) {
                ret = tcp_send(fd, (const char*)BPTR(slot->sq.sb), nbytes);
        } else {
                log_warn("Failed to sent stream for state don't match.\n");
                return;
        }

        if (ret <= 0) {
                slot->io_errno = errnum;
                if (ret == 0) {
                        aio_event_set(&slot->wev, slot->fd, AIO_EV_WRITE, aio_onwrite, slot);
                        aio_evbase_set(&slot->wev, SLOTEBP(slot));
                        if (aio_event_add(&slot->wev, 0) < 0) {
                                log_warn("Failed to add aio event.\n");
                        }
                } else {
                        log_warn("Stream send operate failed.\n");
                        aio_evtimer_add(&slot->eev, &slot->etv);
                }
                return;
        }

        slot->sq.sb->offset += ret;
        slot->sq.sb->length -= ret;

        if (slot->sq.sb->length == 0) {
                AIO_BUF_FREE(slot->sq.sb);
                slot->sq.sb = 0;
                slot->sq.len--;

                if (slot->sq.next) {
                        struct slot_sq *tmpsq = slot->sq.next;
                        slot->sq.sb = tmpsq->sb;
                        slot->sq.next = tmpsq->next;
                        if (slot->sq.tail == tmpsq) {
                                slot->sq.tail = &slot->sq;
                        }
                        free(tmpsq);
                        aio_onwrite(SLOTFD(slot), AIO_EV_WRITE, slot);
                } else {
                        slot->l_write = 0;
                        if (slot->on_sent) (slot->on_sent)(slot);
                }
        } else {
                aio_onwrite(SLOTFD(slot), AIO_EV_WRITE, slot);
        }
}

void
datagram_send(struct aio_slot *slot, struct aio_buf *sb)
{
        struct sockaddr_in addr;
        int ret;

        assert(slot);
        assert(slot->ab);
        assert(sb);

        memset(&addr, 0, sizeof(addr));

        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = slot->dstip;
        addr.sin_port = slot->dstport;

        slot->io_errno = 0;
        ret = sendto(slot->fd, BPTR(sb), BLEN(sb), 0,
                (struct sockaddr *)&addr, sizeof(struct sockaddr));
        if (ret < 0 ) {
                if ((slot->io_errno = errnum) != EWOULDBLOCK) {
			log_warn("Failed to send udp packet, errno: %d\n", errnum);
                        aio_evtimer_add(&slot->eev, &slot->etv);
                }
                AIO_BUF_FREE(sb);
        } else {
                AIO_BUF_FREE(sb);
                if (slot->on_sent) (slot->on_sent)(slot);
        }
}

static int 
datagram_recv(struct aio_slot *slot)
{
        int ret, bytestoread;
        struct sockaddr_in frmaddr;
        int addrlen = sizeof(struct sockaddr_in);

        assert(slot);
        assert(slot->ab);

        slot->rb->offset = AIO_SLOT_RECVBUF_OFFSET;

        bytestoread = MAX_RECVBUF_LEN - BOFF(slot->rb) - BLEN(slot->rb);

        slot->io_errno = 0;
        ret = recvfrom(slot->fd, BEND(slot->rb), bytestoread,
                0, (struct sockaddr *)&frmaddr, (socklen_t*)&addrlen);

        if (ret < 0) {
                slot->io_errno = errnum;
                if (errnum != ECONNRESET)
                {
                        return -1;
                }
                return 0;
        }

        slot->rb->length = ret;
        slot->frmip = frmaddr.sin_addr.s_addr;
        slot->frmport = frmaddr.sin_port;
        return ret;
}

void
stream_send(struct aio_slot *slot, struct aio_buf *sb)
{
        assert(slot);
        assert(slot->ab);
        assert(sb);

        if (slot->sq.len > slot->sqcapability) return;

        if (slot->l_write) {
                struct slot_sq *sq = slot->sq.tail;
                sq->next = (struct slot_sq *)calloc(1, sizeof(struct slot_sq));
                assert(sq->next);
                sq->next->sb = sb;
                slot->sq.len++;
                slot->sq.tail = sq->next;
                if (slot->sq.len > slot->sqcapability) {
                        log_warn("Slot send queue has full %d.\n", slot->sq.len);
                        aio_evtimer_add(&slot->eev, &slot->etv);
                }
        } else {
                slot->l_write = 1;
                slot->sq.len++;
                slot->sq.tail = &slot->sq;
                slot->sq.sb = sb;
                slot->sq.next = 0;
                aio_onwrite(SLOTFD(slot), AIO_EV_WRITE, slot);
        }
}

static int 
stream_recv(struct aio_slot *slot)
{
        int bytestoread, ret;

        assert(slot);
        assert(slot->ab);

#if 0 
        if (slot->bytestoread > MAX_RECVBUF_LEN) {
                log_warn("Max receive buffer length overflow.\n");
                return -1;
        }
#endif

        bytestoread = (slot->bytestoread > 0) ? slot->bytestoread 
                : MAX_RECVBUF_LEN - BOFF(slot->rb) - BLEN(slot->rb);  

        slot->io_errno = 0;

#ifdef _AIO_USE_OPENSSL
        if (slot->type == AIO_TLS && slot->l_connected) {
                ret = ssl_recv(slot->ssl, BEND(slot->rb), bytestoread);
        } else 
#endif /* _AIO_USE_OPENSSL */
        if (slot->type == AIO_FILE || 
                (slot->type == AIO_TCP && slot->l_connected)) {
                ret = tcp_recv(slot->fd, (char*)BEND(slot->rb), bytestoread);
        } else {
                log_warn("Failed to stream received for state don't match.\n");
                return -1;
        }

        if (ret > 0) {
                slot->rb->length += ret;
                if (slot->bytestoread > 0) {
                        slot->bytestoread  -= ret;
                }

                if (slot->bytestoread) { 
                        if (!slot->l_read) {
                                slot->l_read = 1;
                                aio_event_add(&slot->rev, 0);
                        }
                        return 0;
                }
        } else {
                slot->io_errno = errnum;
        }

        return ret;
}

struct aio_slot*
aio_slot_open(struct aio_base *ab, const char *ip,
        unsigned short port, enum slot_type type)
{
        unsigned long tmpip;

        if (ip) {
                tmpip = inet_addr(ip);
        } else {
                tmpip = htonl(INADDR_ANY);
        }

        return aio_slot_open2(ab, tmpip, port, type);
}

struct aio_slot*
aio_slot_open2(struct aio_base *ab, unsigned long ip,
        unsigned short port, enum slot_type type)
{
        int fd, servicetype, prototype = 0, reuse = 0;

        assert(ab);

        if (type < 0) return NULL;

        if (type == AIO_FILE) {
                log_warn("Can't create slot for common file type"
                         "You must be open a file first and assign it's"
                         "fd to aio_slot_open3() function\n");
                return NULL;
        }

        switch(type) {
        case AIO_UDP:
                servicetype = SOCK_DGRAM;
                break;
        case AIO_TCP:
        case AIO_TLS:
                reuse = 1;
                servicetype = SOCK_STREAM;
                break;
        case AIO_ICMP:
                servicetype = SOCK_RAW;
                prototype = IPPROTO_ICMP;
                break;
        case AIO_IGMP:
                servicetype = SOCK_RAW;
                prototype = IPPROTO_IGMP;
                break;
        case AIO_RAW_UDP:
                servicetype = SOCK_RAW;
                prototype = IPPROTO_UDP;
                break;
        case AIO_RAW_TCP:
                servicetype = SOCK_RAW;
                prototype = IPPROTO_TCP;
                break;
        default:
                return NULL;
        }

        fd = create_socket(servicetype, prototype, 1 /* nonblocking */);
        if (fd < 0) {
                log_warn("Create socket faild, errno: %d\n", errnum);
                return NULL;
        }

        if (port > 0) {
                if (bind_socket(fd, ip, port, reuse) != 0) {
                        log_warn("Bind address failed, errno: %d\n", errnum);
                        closesocket(fd);
                        return NULL;
                }
        }

        /* Don't support RAW socket type */
        return aio_slot_open3(ab, fd, ip, port, type);
}

struct aio_slot *
aio_slot_open3(struct aio_base *ab, int fd, unsigned long ip, 
        unsigned short port, enum slot_type type)
{
        struct aio_slot *slot;

        slot = aio_slot_new(ab, fd, 0, 0);
        if (!slot) {
                log_warn("Call aio_slot_new failed.\n");
                return NULL;
        }

        slot->type = type;
        slot->bindip = ip;
        slot->bindport = htons(port);
 
#ifdef _AIO_USE_OPENSSL
        slot->authrequire = SSL_VERIFY_NONE; 
#endif

        LIST_INSERT_HEAD(&ab->slot_head, slot, entries);
        return slot;
}

void
aio_slot_close(struct aio_slot *slot, int fatal)
{
        struct slot_sq *sq;

        if (slot == NULL) return;

        if (--slot->refcnt > 0) {
                debugp("Reduce reference count for aio slot.\n");
                return;
        }

        if (slot->l_close || !slot->ab) 
                return;

        slot->l_close = 1;

        if (slot->sq.sb) {
                AIO_BUF_FREE(slot->sq.sb);
                slot->sq.sb = 0;
        }

        while(slot->sq.next) {
                sq = slot->sq.next;
                slot->sq.next = sq->next;
                if (sq->sb) {
                        AIO_BUF_FREE(sq->sb);
                }
                free(sq);
        }

        if (slot->rb) {
                AIO_BUF_FREE(slot->rb);
                slot->rb = 0;
        }

        if (slot->l_read || slot->l_listening) {
                aio_event_del(&slot->rev);
                slot->l_read = 0;
        }

        if (slot->l_write ||  slot->l_connecting) {
                aio_event_del(&slot->wev);
                slot->l_write = 0;
        }

        if (slot->fd > 0 && fatal) {
                closesocket(slot->fd);
                slot->fd = -1;
        }

#ifdef _AIO_USE_OPENSSL
        if (slot->ssl) {
                SSL_free(slot->ssl);
                slot->ssl = 0;
        }

        if (slot->l_listening) {
                if (slot->ctx) {
                        SSL_CTX_free(slot->ctx);
                        slot->ctx = 0;
                }
        }
#endif /* _AIO_USE_OPENSSL */

        if (slot->on_closed) { 
                (slot->on_closed)(slot);
        }

        LIST_REMOVE(slot, entries);
        free(slot);
}

int 
aio_getttl(struct aio_slot *slot, int *ttl)
{
        assert(slot && ttl);
        return get_socket_ttl(slot->fd, ttl);
}

int
aio_setttl(struct aio_slot *slot, int ttl)
{
        assert(slot);
        return set_socket_ttl(slot->fd, ttl);
}

#ifdef _AIO_USE_OPENSSL
void 
aio_settlsauth(struct aio_slot *slot, unsigned char authtype)
{
        slot->authrequire = authtype;
}
#endif

void 
aio_setsendque(struct aio_slot *slot, int capability)
{
        assert(slot);
        slot->sqcapability = capability;
}

void 
aio_setdstaddr(struct aio_slot *slot, 
        const char *ip, unsigned short port/*host byte order*/)
{
        assert(slot);
        slot->dstip = inet_addr(ip); 
        slot->dstport = htons(port);
}

void 
aio_setdstaddr2(struct aio_slot *slot, 
        unsigned int ip /*network byte order*/, 
        unsigned short port /*host byte order*/)
{
        assert(slot);
        slot->dstip = ip;
        slot->dstport = htons(port);
}

void 
aio_setdstaddr3(struct aio_slot *slot, 
        unsigned int ip/*network byte order*/, 
        unsigned short port/*network byte order*/)
{
        assert(slot);
        slot->dstip = ip;
        slot->dstport = port;
}

int
aio_setread(struct aio_slot *slot, void (*fn)(struct aio_slot *slot), 
        unsigned permanence, unsigned msec)
{
        int ret = 0;
        assert(slot);

        if (!slot->l_read) {
                if (permanence) {
                        aio_event_set(&slot->rev, slot->fd,
                                AIO_EV_READ|AIO_EV_RPERSIST, aio_onread, slot);
                        aio_evbase_set(&slot->rev, SLOTEBP(slot));
                        /* timeout argument is invalid for permance */
                        ret = aio_event_add(&slot->rev, 0);
                } else {
                        aio_event_set(&slot->rev, slot->fd, AIO_EV_READ, aio_onread, slot);
                        aio_evbase_set(&slot->rev, SLOTEBP(slot));
                        if (msec > 0) {
                                struct timeval tv;
                                tv.tv_sec = msec / 1000;
                                tv.tv_usec = (msec % 1000) * 1000; 
                                ret = aio_event_add(&slot->rev, &tv);
                        } else {
                                ret = aio_event_add(&slot->rev, 0);
                        }
                }
        }

        slot->l_read = 1;
        slot->on_received = fn;
        slot->l_permanence = permanence;
 
        return ret;
}

int
aio_settoread(struct aio_slot *slot, unsigned size)
{
        assert(slot);
        assert(slot->ab);

        if (size < 0 || size > MAX_RECVBUF_LEN) 
                return 0;

        slot->bytestoread = size;
        return size;
}

void
aio_setclose(struct aio_slot *slot, void (*fn)(struct aio_slot *slot)) 
{
        assert(slot);
        slot->on_closed = fn;   
}

int  
aio_listen(struct aio_slot *slot, void (*fn)(struct aio_slot *slot))
{
        assert(slot);

        slot->l_listening = 1;
        slot->io_errno = 0;

        if (listen(slot->fd, AIO_DEFAULT_BACKLOG) != 0) {
                slot->io_errno = errnum;
                log_err("Tcp listen failed.\n");
                return -1;
        }

        slot->on_accepted = fn;

        aio_event_set(&slot->rev, slot->fd,
                AIO_EV_READ|AIO_EV_RPERSIST, aio_onaccepted, slot);
        aio_evbase_set(&slot->rev, SLOTEBP(slot));
        if (aio_event_add(&slot->rev, 0) < 0) {
                log_err("Failed to add aio event.\n");
                return -1;
        }
        return 0;
}

#ifdef _AIO_USE_OPENSSL
int
aio_tlslisten(struct aio_slot *slot, void (*fn)(struct aio_slot *slot),
        const char *cacrt, const char *srvcrt, const char *srvkey)
{
        if (ssl_server_init(&slot->ctx, cacrt, srvcrt, srvkey) < 0) {
                log_err("Failed to initialize open ssl context.\n");
                return -1;
        }
        return aio_listen(slot, fn);
}
#endif

int
aio_connect(struct aio_slot *slot,
        void (*fn)(struct aio_slot *slot), unsigned short msec)
{
        int ret;

        assert(slot);

        if (!slot->dstip || !slot->dstport) {
                log_warn("There isn't destination address.\n");
                return -1;
        }

#ifdef _AIO_USE_OPENSSL
        if (slot->type == AIO_TLS) {
                ret = ssl_client_init (slot->fd, &slot->ctx, &slot->ssl);
                if (ret < 0) {
                        log_warn ("Failed to init ssl client.\n");
                        return -1;
                }
        }
#endif
        slot->on_connected = fn;
        slot->l_connecting = 1;

        slot->io_errno = 0;
        ret = tcp_connect(slot->fd, slot->dstip, ntohs(slot->dstport));
        if (ret < 0) {
                log_warn("Failed to establish tcp connection.\n");
                slot->io_errno = errnum;
                return -1;
        } else if (ret == 0) {
                aio_event_set(&slot->wev, slot->fd, AIO_EV_WRITE,
                        aio_onconnected, slot);
                aio_evbase_set(&slot->wev, SLOTEBP(slot));

                if (msec > 0) {
                        struct timeval tv;
                        tv.tv_sec = msec / 1000;
                        tv.tv_usec = (msec % 1000) * 1000; 
                        ret = aio_event_add(&slot->wev, &tv);
                } else {
                        ret = aio_event_add(&slot->wev, 0);
                }

                if (ret < 0) {
                        log_warn("Failed to add aio event.\n");
                        return -1;
                }
                return 0;
        }

        /* Tcp connection succeed. */

#ifdef _AIO_USE_OPENSSL
        if (slot->type == AIO_TLS) {
                aio_tls_connect(slot);
                return 0;
        }
#endif /* _AIO_USE_OPENSSL */

        slot->l_connected = 1;
        if (slot->on_connected) {
                (slot->on_connected)(slot);
        }
        return 0;
}

struct aio_buf*
aio_read(struct aio_slot *slot, int autorm)
{
        struct aio_buf *res = 0;

        assert(slot);
 
        if (slot->rb) 
                res = slot->rb;
 
        if (autorm) 
                slot->rb = 0;
 
        return res;
}  

void
aio_write(struct aio_slot *slot,
        struct aio_buf *sb, void (*fn)(struct aio_slot *slot))
{
        assert(slot);

        if (slot->l_close) {
                AIO_BUF_FREE(sb);
                return;
        }

        slot->on_sent = fn;

        if (slot->type & AIO_STREAM) { /* FILE and TCP and TLS */
                stream_send(slot, sb);
        } else { /* UDP and RAW Sockets */
                datagram_send(slot, sb);
        }
}

